package com.swak.vertx.invoker;

import java.util.concurrent.CompletableFuture;

import com.swak.Constants;
import com.swak.asm.MethodCache.MethodMeta;
import com.swak.reactivex.threads.Context;
import com.swak.reactivex.threads.Contexts;
import com.swak.utils.StringUtils;
import com.swak.vertx.Vertxs;
import com.swak.vertx.transport.codec.Msg;
import com.swak.vertx.transport.script.Script;

import io.vertx.core.AsyncResult;
import io.vertx.core.Handler;
import io.vertx.core.eventbus.Message;

/**
 * 消息的异步处理
 *
 * @author: lifeng
 * @date: 2020/3/29 19:43
 */
public interface FluxInvoker {

	/**
	 * 通过类型解析出服务地址
	 *
	 * @param type 类型
	 * @return 服务地址
	 */
	default String getAddress(Class<?> type, String address) {

		/*
		 * 默认使用接口的全额限定名称
		 */
		if (StringUtils.isBlank(address)) {
			address = type.getName();
		}

		/*
		 * 约定去掉后面的 Async
		 */
		return StringUtils.removeEnd(address, Constants.ASYNC_SUFFIX);
	}

	/**
	 * 消息的异步处理
	 *
	 * @param vertx   vertx 代理
	 * @param address 服务地址
	 * @param method  服务方法
	 * @param args    调用参数
	 * @return 异步结果
	 */
	default CompletableFuture<Object> doRemoteInvoke(String address, Script script) {

		// 构建异步请求结果
		Promise promise = new Promise();

		// 发送消息，处理相应结果
		Vertxs.sentMessage(address, new Msg(script), -1, promise);

		// 返回异步future， futrue收到消息后会触发下一步的操作
		return promise;
	}

	/**
	 * 消息的异步处理
	 *
	 * @param vertx   vertx 代理
	 * @param address 服务地址
	 * @param method  服务方法
	 * @param args    调用参数
	 * @return 异步结果
	 */
	default CompletableFuture<Object> doRemoteInvoke(String address, MethodMeta meta, Object[] args) {

		// 构建异步请求结果
		Promise promise = new Promise();

		// 发送消息，处理相应结果
		Vertxs.sentMessage(address, new Msg(meta, args), meta.getTimeOut(), promise);

		// 返回异步future， futrue收到消息后会触发下一步的操作
		return promise;
	}

	/**
	 * 消息的异步处理
	 *
	 * @param vertx   vertx 代理
	 * @param address 服务地址
	 * @param method  服务方法
	 * @param args    调用参数
	 * @return 异步结果
	 */
	default CompletableFuture<Object> doRemoteInvoke(String address, MethodMeta meta, Object[] args,
			boolean callBackInSourceContext) {

		// 构建异步请求结果
		Promise promise = new Promise(callBackInSourceContext);

		// 发送消息，处理相应结果
		Vertxs.sentMessage(address, new Msg(meta, args), meta.getTimeOut(), promise);

		// 返回异步future， futrue收到消息后会触发下一步的操作
		return promise;
	}

	/**
	 * 减少对象的创建
	 * 
	 * @author DELL
	 */
	static class Promise extends ForbidBlockCompletableFuture<Object> implements Handler<AsyncResult<Message<Msg>>> {
		Context context = null;

		Promise() {
		}

		Promise(boolean callBackInSourceContext) {
			if (callBackInSourceContext) {
				this.context = Contexts.currentContext();
			}
		}

		/**
		 * 这个返回一定会在 Vertx 的上下文中执行, 如果在Context中执行，则进行切换
		 */
		@Override
		public void handle(AsyncResult<Message<Msg>> event) {
			if (context != null) {
				context.runOnContext(() -> this.handleResult(event));
				return;
			}
			this.handleResult(event);
		}

		private void handleResult(AsyncResult<Message<Msg>> event) {

			// 错误 和输出
			Throwable result_error = null;
			Object result_result = null;

			// 内部处理异常
			if (event.cause() != null) {
				result_error = event.cause();
			}

			// 处理正确
			else {
				// 约定的通讯协议
				Msg result = event.result().body();

				// 错误处理 - 结果返回
				result_error = result.getError();
				result_result = result.getResult();

				// 自动生成异步接口返回值
				// if (meta.getNestedReturnType() == Msg.class) {
				// result_result = result;
				// }
			}

			// 优先错误处理
			if (result_error != null) {
				this.completeExceptionally(result_error);
			}

			// 结果返回
			else {
				this.complete(result_result);
			}
		}
	}
}
